#include "connection.h"
#include "connectionMgr.h"

#include <boost/beast/core/buffers_to_string.hpp>
#include <iostream>

#include <boost/uuid/uuid.hpp>
#include <boost/uuid/uuid_io.hpp>
#include <boost/uuid/uuid_generators.hpp>
#include <mutex>

Connection::Connection(boost::asio::io_context& io_context)
:_ws_stream(std::make_unique<stream>(boost::asio::make_strand(io_context))) //保了所有通过 _ws_stream 提交的异步操作都会按照提交顺序执行，从而保证线程安全。
,_io_context(io_context)
{
    boost::uuids::uuid uuid = boost::uuids::random_generator()();
    _uuid = boost::uuids::to_string(uuid);
}

void Connection::start()
{
    auto self = shared_from_this();
    _ws_stream->async_read(_recv_buffer, 
        [self](boost::system::error_code ec, std::size_t bytes_transferred) {
        try 
        {
            if(ec)
            {
                std::cerr << "async_read error: " << ec.message() << std::endl;
                ConnectionMgr::get_instance().rmv_connection(self->uuid());
                return;
            }
            
            self->_ws_stream->text(self->_ws_stream->got_text());
            std::string recv_data = boost::beast::buffers_to_string(self->_recv_buffer.data());
            self->_recv_buffer.consume(self->_recv_buffer.size()); // 清空缓冲区
            std::cout << "recv data: " << recv_data << std::endl;

            self->async_send(std::move(recv_data));
            self->start();
        } 
        catch (std::exception& e) 
        {
            std::cerr << "async_read error: " << e.what() << std::endl;
            ConnectionMgr::get_instance().rmv_connection(self->uuid());
        }   
    });
}

void Connection::async_accept()
{
    auto self = shared_from_this();
    _ws_stream->async_accept(
        [self](boost::system::error_code ec) {
        try 
        {
            if(ec)
            {
                std::cerr << "async_accept error: " << ec.message() << std::endl;
                return;
            }
            
            ConnectionMgr::get_instance().add_connection(self);
            self->start();
        } 
        catch (std::exception& e) 
        {
            std::cerr << "async_accept error: " << e.what() << std::endl;
        }
    });

}

void Connection::async_send(std::string&& data)
{
    {
        std::unique_lock<std::mutex> lock(_send_queue_lock);
        auto len = _send_queue.size();
        _send_queue.emplace(data);

        if(len > 0)
        {
            return;
        }
    }

    auto self = shared_from_this();
    _ws_stream->async_write(boost::asio::buffer(data.c_str(), data.size()),
        [self](boost::system::error_code ec, std::size_t bytes_transferred) {
        try 
        {
            if(ec)
            {
                std::cerr << "async_write error: " << ec.message() << std::endl;
                ConnectionMgr::get_instance().add_connection(self);
                return;
            }
            
            std::string send_msg;
            {
                std::unique_lock<std::mutex> lock(self->_send_queue_lock);
                self->_send_queue.pop();
                if(self->_send_queue.empty())
                {
                    return;
                }

                send_msg = self->_send_queue.front();
            }

            self->async_send(std::move(send_msg));
        } 
        catch (std::exception& e) 
        {
            ConnectionMgr::get_instance().add_connection(self);
            std::cerr << "async_write error: " << e.what() << std::endl;
        }
    });
}

void Connection::close()
{
    socket().close();
}